package cn.com.ytst.data.integration.components.reader.kafka;

import java.io.File;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import com.alibaba.datax.common.util.Configuration;
import cn.com.ytst.data.integration.core.plugin.Reader;
import cn.com.ytst.data.integration.util.Constants;

/**
 * @ClassName KafkaReader
 * @Deacription kafka 读取数据
 * @Author Administrator
 * @Date 2020/1/1 0001 19:49
 * @Version 1.0
 **/
public class KafkaReader extends Reader {


    Properties properties = new Properties();
    private static final String LINE_TAG = File.separator;

    protected String bootstrapServers;

    protected File file;

    protected String topic;

    protected String key;

    protected Configuration configuration;

    protected String maxRecords;

    protected MqPrarmsVo mqPrarmsVo;

    protected KafkaConsumer<String,String> consumer;

    public KafkaReader(File file) {


    }






    @Override
    public void init() {
        this.topic = configuration.getString(Constants.KAFAK_WRITER_TOPIC_TAG);
        this.bootstrapServers = configuration.getString(Constants.KAFAK_WRITER_BOOTSTRAPSERVERS_TAG);
        this.key = configuration.getString(Constants.KAFAK_WRITER_KEY_TAG);
        properties.put("bootstrap.servers", this.bootstrapServers);
        properties.put("max.poll.records", 1);
        properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 50000);
        properties.put("enable.auto.commit", false);
        properties.put("group.id", topic);
        properties.put("auto.offset.reset", false);
        // props.put("max.poll.interval.ms", value)
        properties.put("acks", "1");
        properties.put("session.timeout.ms", 200);
        properties.put("key.serializer", KafkaWriteKey.KAFKA_PRODUCER_KEY_SERIALIZER);
        properties.put("value.serializer", KafkaWriteKey.KAFKA_PRODUCER_VALUE_SERIALIZER);
        consumer = new  KafkaConsumer<>(properties);
    }

    @Override
    public void startReader() {

    }

    @Override
    public void destroy() {

    }

    private MqPrarmsVo builder(File file ) {
        Configuration configuration = Configuration.from(file);
        String host = configuration.getString(Constants.MQ_HOST_TAG);
        //用户名或者密码有可能存在加密
        String username = configuration.getString(Constants.MQ_USERNAME_TAG,Constants.DEFAULT_VALUE);
        String password = configuration.getString(Constants.MQ_PWD_TAG,Constants.DEFAULT_VALUE);
        String port = configuration.getString(Constants.MQ_PORT_TAG,Constants.DEFAULT_VALUE);
        boolean encrypt = configuration.getBool(Constants.MQ_ENCRYPT_TAG,false);
        String queue = configuration.getString(Constants.MQ_QUEUE_TAG,Constants.DEFAULT_VALUE);
        String topic = configuration.getString(Constants.TOPIC_TAG,Constants.DEFAULT_VALUE);
        String key = configuration.getString(Constants.KEY_TAG,Constants.DEFAULT_VALUE);
        String schema = configuration.getString(Constants.MQ_SCHEMA);
        //String type = configuration.getString(Constants.MQ_TYPE_TAG);//消息中间件类型
        String sysName = configuration.getString(Constants.MQ_SYSNAME);
        String tableName = configuration.getString(Constants.MQ_TABLE_NAME);
        String airport = configuration.getString(Constants.MQ_AIRPORT);
        boolean isBatch = configuration.getBool(Constants.MQ_IS_BATCH,true);
        int pageSize = configuration.getInt(Constants.MQ_PAGESIZE,100);
        SimpleDateFormat sf = new SimpleDateFormat("yyyyMMddHHmmss");

        MqPrarmsVo vo = new MqPrarmsVo(username, password, host, port, queue, topic,
                key, schema, sysName, tableName, airport, isBatch, pageSize,sf.format(new Date()),configuration);
        String path = file.getPath();
        int last_index = (path).lastIndexOf(LINE_TAG);
        vo.setFilename((path).substring(last_index+1, path.length()));
        vo.setEncrypt(encrypt);
        return vo;
    }
}
